package com.rock.code.province.tj.busi.socketdr;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.logging.Logger;

/**
 * 生产者
 *
 * @author sunhaobo
 * @date 2019 /09/10
 */
public class Producer {
   private Logger logger = Logger.getLogger(Producer.class.getName());
    /**
     * The Producer.
     */
    private final KafkaProducer<String, String> producer;
    /**
     * The constant TOPIC.
     */
    public final static String TOPIC = "test";  //创建kafka主题

    /**
     * Instantiates a new Producer.
     *
     * @author sunhaobo
     * @date 2019 /09/10
     */
    //初始化同类名构造方法，New类到时候，会初始化
    public Producer(){
        Properties props = new Properties();
        // 服务器ip
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.43.74:9092");
        // 属性键值对都序列化成字符串
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        // 创建一个生产者
        producer = new KafkaProducer<String, String>(props);
    }

    /**
     * Produce.
     *
     * @author sunhaobo
     * @date 2019 /09/10
     */
    public void produce(String data) {
        int messageNo = 1;
        final int COUNT = 2;
        while (messageNo < COUNT) {
            String key = String.valueOf(messageNo);
//             data = "sunhaobo hello kafka message " + key;
            boolean sync = false;   //是否同步
            if (sync) {
                try {
                    producer.send(new ProducerRecord<String, String>(TOPIC, key,data)).get();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } else {
                producer.send(new ProducerRecord<String, String>(TOPIC, key,data));
                logger.info(data);

            }
            //必须写下面这句,相当于发送
            producer.flush();
            messageNo ++;
        }
    }

    /**
     * The entry point of application.
     *
     * @param args the input arguments
     * @author sunhaobo
     * @date 2019 /09/10
     */
//    public static void main( String[] args ) {
//        new Producer().produce();
//    }
}